package com.example;

import com.example.utils.ConnectionFactoryUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;

import java.util.HashMap;
import java.util.Map;

/**
 * @author wangxiyuan
 * @description
 * @createDate 2021/3/24 11:46
 **/
public class Worker {
    private static final String TASK_QUEUE_NAME = "task_queue";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = ConnectionFactoryUtils.getConnectionFactory();
        final Connection connection = factory.newConnection();
        final Channel channel = connection.createChannel();

        channel.queueDeclare(TASK_QUEUE_NAME, false, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
        // 值0被视为无穷大，允许任何数量的未确认消息。
        // 该值定义通道上允许的未确认交付的最大数量。
        channel.basicQos(10);
        // 默认为 false,表示针对每个用户 进行限制
        // 当设置为true时，表示针对每个通道进行限制，也就是通道上的所有用户共享未确认消息量
        //channel.basicQos(1,true);

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");

            System.out.println(" [x] Received '" + message + "'");
            try {
                doWork(message);
            } finally {
                System.out.println(" [x] Done");
                // positively acknowledge a single delivery, the message will be discarded
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        };
        // 设置消费端的优先级
        Map<String, Object> args = new HashMap<String, Object>();
        args.put("x-priority", 10);
        /*
            第一个参数：队列名称
            第二个参数：设置是否自动确认模式
         */

        String ctag = channel.basicConsume(TASK_QUEUE_NAME, false,args, deliverCallback, consumerTag -> {
        });
       // channel.basicCancel(ctag);
    }

    private static void doWork(String task) {
        for (char ch : task.toCharArray()) {
            if (ch == '.') {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException _ignored) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }
}
